package com.example.zz.example.pattern.rxjava;

import android.os.Bundle;
import android.util.Log;
import android.widget.Toast;

import androidx.appcompat.app.AppCompatActivity;

import com.example.zz.example.R;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * 简单入门例子：https://www.jianshu.com/p/fce825833d36
 * 全面讲解： https://gank.io/post/560e15be2dca930e00da1083
 * <p>
 * RxJava2.0 使用
 * Observable：被观察者
 * Observer：观察者
 * subscribe：订阅动作
 * 使用时，新建被观察者对象和观察者对象，通过subscribe将两个对象连接，之后就可以通过Observable对象发送事件给subscribe对象接受
 * <p>
 * RxJava2.x 有两大类使用方式
 * 1、Observable - subscribe - Observer     不支持背压
 * 2、Flowable - subscribe - Subscriber      支持背压
 * <p>
 * RxJava1.x 观察者都一样，现在不推荐使用了
 * Observable - subscribe - Observer
 * Observable - subscribe - Subscriber
 * 背压：在异步场景中，被观察者发送事件的速度远快于观察者处理事件的速度情况下，一种告诉上游被观察者降低发送速度的策略
 */

public class RxjavaActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_rxjava);

        RxJavaExample rxJavaExample = new RxJavaExample();
        findViewById(R.id.order).setOnClickListener(v -> {
            rxJavaExample.startExample(RxjavaActivity.this);
            rxJavaExample.testDoOnOrder();
            rxJavaExample.threadChangeExample();
        });
        findViewById(R.id.observer).setOnClickListener(v -> {
            // 1、Observable - subscribe - Observer     不支持背压
            observerEvent();
        });
        findViewById(R.id.subscriber).setOnClickListener(v -> {
            // 2、Flowable - subscribe - Subscriber      支持背压
            subscriberEvent();
        });
        findViewById(R.id.observer1).setOnClickListener(v -> {
            observerEvent1();
        });
        findViewById(R.id.observer2).setOnClickListener(v -> {
            observerEvent2();
        });
        findViewById(R.id.observer3).setOnClickListener(v -> {
            observerEvent3();
        });

    }

    private void observerEvent() {
        //基础写法分三步
        //第一步：创建被观察者，注意创建的泛型里面传入的类型
        Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) {
                for (int i = 0; i < 3; i++) {
                    emitter.onNext(i);
                }
                emitter.onComplete();
            }
        });

        //第二步：创建观察者， 注意泛型中的类型
        Observer<Integer> observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Toast.makeText(RxjavaActivity.this, "result : " + integer, Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onError(Throwable e) {
                Toast.makeText(RxjavaActivity.this, "onError : " + e.getMessage(), Toast.LENGTH_SHORT).show();
            }

            @Override
            public void onComplete() {
                Toast.makeText(RxjavaActivity.this, "onComplete", Toast.LENGTH_SHORT).show();
            }
        };

        //第三步：将观察者和被观察者联系起来，调用此代码后，就会触发observable中的subscriber发送事件，在observer中的onNext接收
        observable.subscribe(observer);


        //简化写法1 将上面的代码使用链式方式调用+lambda简化
        Observable.create((ObservableOnSubscribe<String>) emitter -> {
                    emitter.onNext("111");
                    emitter.onNext("222");
                    emitter.onComplete();
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {

                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

//        subscribe有多种复写方法及变种写法，适合不同场景。
//        subscribe() 所有结果都不观察
//        subscribe(Observer observer) observer中观察所有结果，其它的重载方法都是部分实现observer的方法
//        subscribe(Consumer onNext) 只观察onNext中的数据

        //简化写法2 不关心发送后的数据结果
        Observable.create((ObservableOnSubscribe<String>) emitter -> {
                    emitter.onNext("111");
                    emitter.onNext("222");
                    emitter.onComplete();
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe();

        //简化写法3 只关心onNext数据
        Disposable subscribe = Observable.create((ObservableOnSubscribe<String>) emitter -> {
                    emitter.onNext("11");
                    emitter.onNext("22");
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Toast.makeText(RxjavaActivity.this, "accept : " + s, Toast.LENGTH_SHORT).show();
                    }
                });
    }

    /**
     * 上游可以发送无限个onNext, 下游也可以接收无限个onNext.
     * 当上游发送了一个onComplete后, 上游onComplete之后的事件将会继续发送, 而下游收到onComplete事件之后将不再继续接收事件.
     * 当上游发送了一个onError后, 上游onError之后的事件将继续发送, 而下游收到onError事件之后将不再继续接收事件.
     * 上游可以不发送onComplete或onError.如果不发的话，观察者一直处于观察状态，其它操作符可能受影响！！！ 如repeat
     * 最为关键的是onComplete和onError必须唯一并且互斥, 即不能发多个onComplete, 也不能发多个onError, 也不能先发一个onComplete,
     * 然后再发一个onError, 反之亦然。
     *
     * 上面的几条都好理解，关键是最后一条，这是需要我们在代码中控制的，虽然违反最后一条并不一定导致程序崩溃。
     * 比如发送多个onComplete是可以正常运行的, 依然是收到第一个onComplete就不再接收了, 但若是发送多个onError,
     * 则收到第二个onError事件会导致程序会崩溃。

     * repeat：重复此操作符上面的Observable内容，所以位置很重要，不同位置Observable有区别，结果也有变化;当repeat() 接收到onComplete()会触发重订阅
     * <p>
     * 案例1，希望emitter.onNext调用两次后结束
     */
    private void observerEvent1() {
        Log.e("eeeeee", "案例1 ===========");
        Observable.create((ObservableOnSubscribe<String>) emitter -> {
                    String s = "aaa";
                    Log.e("eeeeee", s);
                    emitter.onNext(s + "bbb");
                    emitter.onComplete();
                })
                .repeat(2)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(String s) {
                        Log.e("eeeeee", "onNext = " + s);
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onComplete() {
                        Log.e("eeeeee", "onComplete ");
                    }
                });
        //2022-10-10 16:20:48.497 4206-4277/com.example.zz.example E/eeeeee: aaa
        //2022-10-10 16:20:48.497 4206-4277/com.example.zz.example E/eeeeee: aaa
        //2022-10-10 16:20:48.502 4206-4206/com.example.zz.example E/eeeeee: onNext = aaabbb
        //2022-10-10 16:20:48.502 4206-4206/com.example.zz.example E/eeeeee: onNext = aaabbb
        //2022-10-10 16:20:48.502 4206-4206/com.example.zz.example E/eeeeee: onComplete
        //repeat在emitter ->{}之后，在subscribeOn(Schedulers.io())之前，所以重复的是emitter ->{}中的非onComplete内容，
        // 所以在4206-4277线程中先后两次打印aaa并发射，在主线程接收后打印aaabbb，最后onComplete
    }

    /**
     * 案例2，希望emitter.onNext调用两次后结束，但是没有调用onComplete，导致repeat失效
     */
    private void observerEvent2() {
        Log.e("eeeeee", "案例2 ===========");
        Observable.create((ObservableOnSubscribe<String>) emitter -> {
                    String s = "aaa";
                    Log.e("eeeeee", s);
                    emitter.onNext(s + "bbb");
                })
                .repeat(2)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(String s) {
                        Log.e("eeeeee", "onNext = " + s);
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onComplete() {
                        Log.e("eeeeee", "onComplete ");
                    }
                });
        //2022-10-10 16:44:11.339 4795-4847/com.example.zz.example E/eeeeee: aaa
        //2022-10-10 16:44:11.342 4795-4795/com.example.zz.example E/eeeeee: onNext = aaabbb
        //没有重复打印，repeat(2)失效了 ！！！，流程只进行了一次
    }

    /**
     * 案例3，repeat位置不同时，结果不一样！
     */
    private void observerEvent3() {
        Log.e("eeeeee", "案例3 ===========");
        Observable.create((ObservableOnSubscribe<String>) emitter -> {
                    String s = "aaa";
                    Log.e("eeeeee", s);
                    emitter.onNext(s + "bbb");
                    emitter.onComplete();
                })
                .subscribeOn(Schedulers.io())
                .repeat(2)
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                    }

                    @Override
                    public void onNext(String s) {
                        Log.e("eeeeee", "onNext = " + s);
                    }

                    @Override
                    public void onError(Throwable e) {
                    }

                    @Override
                    public void onComplete() {
                        Log.e("eeeeee", "onComplete ");
                    }
                });
//        打印结果
//        2022-10-10 16:48:42.892 5068-5120/com.example.zz.example E/eeeeee: aaa
//        2022-10-10 16:48:42.893 5068-5068/com.example.zz.example E/eeeeee: onNext = aaabbb
//        2022-10-10 16:48:42.893 5068-5121/com.example.zz.example E/eeeeee: aaa
//        2022-10-10 16:48:42.893 5068-5068/com.example.zz.example E/eeeeee: onNext = aaabbb
//        2022-10-10 16:48:42.893 5068-5068/com.example.zz.example E/eeeeee: onComplete
//        或者出现
//        2022-10-10 17:00:26.219 5585-5630/com.example.zz.example E/eeeeee: aaa
//        2022-10-10 17:00:26.219 5585-5631/com.example.zz.example E/eeeeee: aaa
//        2022-10-10 17:00:26.226 5585-5585/com.example.zz.example E/eeeeee: onNext = aaabbb
//        2022-10-10 17:00:26.226 5585-5585/com.example.zz.example E/eeeeee: onNext = aaabbb
//        2022-10-10 17:00:26.226 5585-5585/com.example.zz.example E/eeeeee: onComplete
//        对比案例1，repeat移动到subscribeOn(Schedulers.io())的下面，打印效果出现差异
//        由于repeat重复之前内容的特性，所以第二次时，重新再另一个io线程中完成的流程，两次线程5120和5121；案例1的都在同一子线程


//        如果将repeat(2)再向下移动到observeOn(AndroidSchedulers.mainThread())下面，aaa的打印又是同一个线程了
//        2022-10-10 16:57:51.934 5336-5400/com.example.zz.example E/eeeeee: aaa
//        2022-10-10 16:57:51.934 5336-5336/com.example.zz.example E/eeeeee: onNext = aaabbb
//        2022-10-10 16:57:51.934 5336-5400/com.example.zz.example E/eeeeee: aaa
//        2022-10-10 16:57:51.934 5336-5336/com.example.zz.example E/eeeeee: onNext = aaabbb
//        2022-10-10 16:57:51.934 5336-5336/com.example.zz.example E/eeeeee: onComplete

    }

    /**
     * 背压写法和Observable类似，区别是字段及参数，背压需要添加背压策略
     * <p>
     * 上游发射数据的速度快于下游接收处理数据的速度，这样对于那些没来得及处理的数据就会造成积压，
     * 这些数据既不会丢失，也不会被垃圾回收机制回收，而是存放在一个异步缓存池中，如果缓存池中的数据一直得不到处理，
     * 越积越多，最后就会造成内存溢出，这便是响应式编程中的背压（backpressure）问题
     * <p>
     * BackpressureStrategy背压策略
     * <p>
     * MISSING（MissingEmitter）：在此策略下，通过Create方法创建的Flowable相当于没有指定背压策略，不会对通过onNext发射的数据做缓存或丢弃处理，需要下游通过背压操作符。
     * ERROR（ErrorAsyncEmitter）：在此策略下，如果放入Flowable的异步缓存池中的数据超限了，则会抛出MissingBackpressureException异常。
     * BUFFER（BufferAsyncEmitter）：部维护了一个缓存池SpscLinkedArrayQueue，其大小不限，此策略下，如果Flowable默认的异步缓存池满了，会通过此缓存池暂存数据，它与Observable的异步缓存池一样，可以无限制向里添加数据，不会抛出MissingBackpressureException异常，但会导致OOM。
     * DROP（DropAsyncEmitter）：在此策略下，如果Flowable的异步缓存池满了，会丢掉上游发送的数据
     * LATEST（LatestAsyncEmitter）：与Drop策略一样，如果缓存池满了，会丢掉将要放入缓存池中的数据，不同的是，不管缓存池的状态如何，LATEST都会将最后一条数据强行放入缓存池中，来保证观察者在接收到完成通知之前，能够接收到Flowable最新发射的一条数据
     * <p>
     * 原文链接：https://blog.csdn.net/weixin_43506260/article/details/105037209
     */
    private void subscriberEvent() {

        //第一步，创建Flowable
        Flowable<Object> objectFlowable = Flowable.create(new FlowableOnSubscribe<Object>() {
            @Override
            public void subscribe(FlowableEmitter<Object> emitter) throws Exception {
                emitter.onNext("Flowable-luo");
                emitter.onComplete();
            }
        }, BackpressureStrategy.BUFFER);

        //第二布，创建subscriber
        Subscriber<Object> subscriber = new Subscriber<Object>() {

            @Override
            public void onSubscribe(Subscription s) {
                s.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Object o) {

            }

            @Override
            public void onError(Throwable t) {

            }

            @Override
            public void onComplete() {

            }
        };
        //第三步，绑定观察者
        objectFlowable.subscribe(subscriber);


    }
}